package org.huel.dataprocessing

import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SQLContext}

import java.util
import java.util.concurrent.{Executors, TimeUnit}
import java.util.{Arrays, Properties}
import scala.jdk.CollectionConverters.IterableHasAsScala

/**
 * @Description Kafka 数据处理，统计考勤并更新到 MySQL
 * @Author Liweijian
 * @Date 2024/11/28
 */
object runSQL {

  private case class Attendance(course: String, name: String, studentId: String, status: String)

  def main(args: Array[String]): Unit = {
    // 创建 SparkConf
    val conf = new SparkConf().setAppName("SparkSql").setMaster("local[*]")
    val sc = new SparkContext(conf)
    val sqlCon = new SQLContext(sc)
    sc.setLogLevel("error")

    // Kafka 配置信息
    val properties: Properties = new Properties
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "43.143.125.94:9092")
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "sparksql")
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

    // 构建 Kafka 消费者
    val kfkconsumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](properties)
    kfkconsumer.subscribe(util.Arrays.asList("attendance"))

    // 创建一个调度器
    val scheduler = Executors.newScheduledThreadPool(1)

    // 定义一个任务，每隔10秒执行一次
    val task = new Runnable {
      def run(): Unit = {
        // 拉取数据
        val records = kfkconsumer.poll(1000).asScala
        val data = records
          .map(record => record.value().split("\t"))
          .map(arr => Attendance(arr(2), arr(1), arr(3), arr(4)))
          .toList

        if (data.nonEmpty) {
          // 将数据转换为 RDD 和 DataFrame
          val attendanceRDD = sc.parallelize(data)
          import sqlCon.implicits._
          val attendanceDF = attendanceRDD.toDF()

          // 注册临时表
          attendanceDF.createOrReplaceTempView("attendance")

          // SQL 查询以统计结果
          val statsDF = sqlCon.sql("""
            SELECT course,
                   COUNT(CASE WHEN status = 'L' THEN 1 END) AS lCount,
                   COUNT(CASE WHEN status = 'A' THEN 1 END) AS aCount
            FROM attendance
            GROUP BY course
          """)

          // 显示统计结果
          statsDF.show()

          // 定义 MySQL 连接属性
          val jdbcUrl = "jdbc:mysql://43.143.125.94:3306/attendance?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai"
          val dbTable = "LAByCourse"
          val connectionProperties = new Properties()
          connectionProperties.put("user", "root")
          connectionProperties.put("password", "Lwj378$$")

          // 执行累加更新操作
          statsDF.foreachPartition { (partition: Iterator[Row]) =>
            val connection = java.sql.DriverManager.getConnection(jdbcUrl, connectionProperties)
            try {
              val statement = connection.createStatement()
              try {
                partition.foreach { row =>
                  val course = row.getString(0)
                  val newLCount = row.getLong(1)
                  val newACount = row.getLong(2)

                  // 查询当前数据库中已有的数据
                  val querySql = s"SELECT lCount, aCount FROM $dbTable WHERE course = '$course'"
                  val rs = statement.executeQuery(querySql)

                  var currentLCount = 0L
                  var currentACount = 0L
                  if (rs.next()) {
                    currentLCount = rs.getLong("lCount")
                    currentACount = rs.getLong("aCount")
                  }
                  rs.close()

                  // 计算累加后的值
                  val updatedLCount = currentLCount + newLCount
                  val updatedACount = currentACount + newACount

                  // 更新或插入数据库
                  val upsertSql = s"""
                    INSERT INTO $dbTable (course, lCount, aCount)
                    VALUES ('$course', $updatedLCount, $updatedACount)
                    ON DUPLICATE KEY UPDATE
                      lCount = $updatedLCount,
                      aCount = $updatedACount
                  """
                  statement.executeUpdate(upsertSql)
                }
              } finally {
                statement.close()
              }
            } finally {
              connection.close()
            }
          }

          println("Data processed and stored with accumulation")
        } else {
          println("No data")
        }
      }
    }

    // 安排任务，每隔5秒执行一次
    scheduler.scheduleAtFixedRate(task, 0, 5, TimeUnit.SECONDS)
  }
}
